/**
 * @author Zhaonian Zou <znzou@hit.edu.cn>,
 * School of Computer Science and Technology,
 * Harbin Institute of Technology, China
 */

#include "executor.h"

#include <exceptions/buffer_exceeded_exception.h>
#include <cmath>
#include <ctime>
#include <functional>
#include <iostream>
#include <string>
#include <utility>

#include "file_iterator.h"
#include "page_iterator.h"
#include "storage.h"

using namespace std;

namespace badgerdb {

    void TableScanner::print() const {
        badgerdb::File file = badgerdb::File::open(tableFile.filename());
        for (badgerdb::FileIterator iter = file.begin(); iter != file.end(); ++iter) {
            badgerdb::Page page = *iter;
            badgerdb::Page *buffered_page;
            bufMgr->readPage(&file, page.page_number(), buffered_page);
            for (badgerdb::PageIterator page_iter = buffered_page->begin();
                 page_iter != buffered_page->end(); ++page_iter) {
                string key = *page_iter;
                string print_key = "(";
                int current_index = 0;
                for (int i = 0; i < tableSchema.getAttrCount(); ++i) {
                    switch (tableSchema.getAttrType(i)) {
                        case INT: {
                            int true_value = 0;
                            for (int j = 0; j < 4; ++j) {
                                if (std::string(key, current_index + j, 1)[0] == '\0') {
                                    continue;  // \0 is actually representing 0
                                }
                                true_value +=
                                        (std::string(key, current_index + j, 1))[0] * pow(128, 3 - j);
                            }
                            print_key += to_string(true_value);
                            current_index += 4;
                            break;
                        }
                        case CHAR: {
                            int max_len = tableSchema.getAttrMaxSize(i);
                            print_key += std::string(key, current_index, max_len);
                            current_index += max_len;
                            current_index +=
                                    (4 - (max_len % 4)) % 4;  // align to the multiple of 4
                            break;
                        }
                        case VARCHAR: {
                            int actual_len = key[current_index];
                            current_index++;
                            print_key += std::string(key, current_index, actual_len);
                            current_index += actual_len;
                            current_index +=
                                    (4 - ((actual_len + 1) % 4)) % 4;  // align to the multiple of 4
                            break;
                        }
                    }
                    print_key += ",";
                }
                print_key[print_key.size() - 1] = ')';  // change the last ',' to ')'
                cout << print_key << endl;
            }
            bufMgr->unPinPage(&file, page.page_number(), false);
        }
        bufMgr->flushFile(&file);
    }

    JoinOperator::JoinOperator(File &leftTableFile,
                               File &rightTableFile,
                               const TableSchema &leftTableSchema,
                               const TableSchema &rightTableSchema,
                               const Catalog *catalog,
                               BufMgr *bufMgr)
            : leftTableFile(leftTableFile),
              rightTableFile(rightTableFile),
              leftTableSchema(leftTableSchema),
              rightTableSchema(rightTableSchema),
              resultTableSchema(
                      createResultTableSchema(leftTableSchema, rightTableSchema)),
              catalog(catalog),
              bufMgr(bufMgr),
              isComplete(false) {
        // nothing
    }

    TableSchema JoinOperator::createResultTableSchema(
            const TableSchema &leftTableSchema,
            const TableSchema &rightTableSchema) {
        vector <Attribute> attrs;

        // first add all the left table attrs to the result table
        for (int k = 0; k < leftTableSchema.getAttrCount(); ++k) {
            Attribute new_attr = Attribute(
                    leftTableSchema.getAttrName(k), leftTableSchema.getAttrType(k),
                    leftTableSchema.getAttrMaxSize(k), leftTableSchema.isAttrNotNull(k),
                    leftTableSchema.isAttrUnique(k));
            attrs.push_back(new_attr);
        }

        // test every right table attrs, if it doesn't have the same attr(name and
        // type) in the left table, then add it to the result table
        for (int i = 0; i < rightTableSchema.getAttrCount(); ++i) {
            bool has_same = false;
            for (int j = 0; j < leftTableSchema.getAttrCount(); ++j) {
                if ((leftTableSchema.getAttrType(j) == rightTableSchema.getAttrType(i)) &&
                    (leftTableSchema.getAttrName(j) == rightTableSchema.getAttrName(i))) {
                    has_same = true;
                }
            }
            if (!has_same) {
                Attribute new_attr = Attribute(
                        rightTableSchema.getAttrName(i), rightTableSchema.getAttrType(i),
                        rightTableSchema.getAttrMaxSize(i), rightTableSchema.isAttrNotNull(i),
                        rightTableSchema.isAttrUnique(i));
                attrs.push_back(new_attr);
            }
        }
        return TableSchema("TEMP_TABLE", attrs, true);
    }

    void JoinOperator::printRunningStats() const {
        cout << "# Result Tuples: " << numResultTuples << endl;
        cout << "# Used Buffer Pages: " << numUsedBufPages << endl;
        cout << "# I/Os: " << numIOs << endl;
    }

    vector <Attribute> JoinOperator::getCommonAttributes(
            const TableSchema &leftTableSchema,
            const TableSchema &rightTableSchema) const {
        vector <Attribute> common_attrs;
        for (int i = 0; i < rightTableSchema.getAttrCount(); ++i) {
            for (int j = 0; j < leftTableSchema.getAttrCount(); ++j) {
                if ((leftTableSchema.getAttrType(j) == rightTableSchema.getAttrType(i)) &&
                    (leftTableSchema.getAttrName(j) == rightTableSchema.getAttrName(i))) {
                    Attribute new_attr = Attribute(rightTableSchema.getAttrName(i),
                                                   rightTableSchema.getAttrType(i),
                                                   rightTableSchema.getAttrMaxSize(i),
                                                   rightTableSchema.isAttrNotNull(i),
                                                   rightTableSchema.isAttrUnique(i));
                    common_attrs.push_back(new_attr);
                }
            }
        }
        return common_attrs;
    }

/**
 * use the original key to generate the search key
 * @param key
 * @param common_attrs
 * @param TableSchema
 * @return
 */
    string construct_search_key(string key,
                                vector <Attribute> &common_attrs,
                                const TableSchema &TableSchema) {
        string search_key;
        int current_index = 0;
        int current_attr_index = 0;
        for (int i = 0; i < TableSchema.getAttrCount(); ++i) {
            switch (TableSchema.getAttrType(i)) {
                case INT: {
                    if (TableSchema.getAttrName(i) ==
                        common_attrs[current_attr_index].attrName &&
                        TableSchema.getAttrType(i) ==
                        common_attrs[current_attr_index].attrType) {
                        search_key += std::string(key, current_index, 4);
                        current_attr_index++;
                    }
                    current_index += 4;
                    break;
                }
                case CHAR: {
                    int max_len = TableSchema.getAttrMaxSize(i);
                    if (TableSchema.getAttrName(i) ==
                        common_attrs[current_attr_index].attrName &&
                        TableSchema.getAttrType(i) ==
                        common_attrs[current_attr_index].attrType) {
                        search_key += std::string(key, current_index, max_len);
                        current_attr_index++;
                    }
                    current_index += max_len;
                    current_index += (4 - (max_len % 4)) % 4;;  // align to the multiple of 4
                    break;
                }
                case VARCHAR: {
                    int actual_len = key[current_index];
                    current_index++;
                    if (TableSchema.getAttrName(i) ==
                        common_attrs[current_attr_index].attrName &&
                        TableSchema.getAttrType(i) ==
                        common_attrs[current_attr_index].attrType) {
                        search_key += std::string(key, current_index, actual_len);
                        current_attr_index++;
                    }
                    current_index += actual_len;
                    current_index +=
                            (4 - ((actual_len + 1) % 4)) % 4;  // align to the multiple of 4
                    break;
                }
            }
            if (current_attr_index >= common_attrs.size())
                break;
        }
        return search_key;
    }

    string JoinOperator::joinTuples(string leftTuple,
                                    string rightTuple,
                                    const TableSchema &leftTableSchema,
                                    const TableSchema &rightTableSchema) const {
        int cur_right_index = 0;  // current substring index in the right table key
        string result_tuple = leftTuple;

        for (int i = 0; i < rightTableSchema.getAttrCount(); ++i) {
            bool has_same = false;
            for (int j = 0; j < leftTableSchema.getAttrCount(); ++j) {
                if ((leftTableSchema.getAttrType(j) == rightTableSchema.getAttrType(i)) &&
                    (leftTableSchema.getAttrName(j) == rightTableSchema.getAttrName(i))) {
                    has_same = true;
                }
            }
            // if the key is only owned by right table, add it to the result tuple
            switch (rightTableSchema.getAttrType(i)) {
                case INT: {
                    if (!has_same) {
                        result_tuple += std::string(rightTuple, cur_right_index, 4);
                    }
                    cur_right_index += 4;
                    break;
                }
                case CHAR: {
                    int max_len = rightTableSchema.getAttrMaxSize(i);
                    if (!has_same) {
                        result_tuple += std::string(rightTuple, cur_right_index, max_len);
                    }
                    cur_right_index += max_len;
                    unsigned align_ = (4 - (max_len % 4)) % 4;  // align to the multiple of
                    // 4
                    for (int k = 0; k < align_; ++k) {
                        result_tuple += "0";
                        cur_right_index++;
                    }
                    break;
                }
                case VARCHAR: {
                    int actual_len = rightTuple[cur_right_index];
                    result_tuple += std::string(rightTuple, cur_right_index, 1);
                    cur_right_index++;
                    if (!has_same) {
                        result_tuple += std::string(rightTuple, cur_right_index, actual_len);
                    }
                    cur_right_index += actual_len;
                    unsigned align_ =
                            (4 - ((actual_len + 1) % 4)) % 4;  // align to the multiple of 4
                    for (int k = 0; k < align_; ++k) {
                        result_tuple += "0";
                        cur_right_index++;
                    }
                    break;
                }
            }
        }
        return result_tuple;
    }

    bool OnePassJoinOperator::execute(int numAvailableBufPages, File &resultFile) {
        if (isComplete)
            return true;

        numResultTuples = 0;
        numUsedBufPages = 0;
        numIOs = 0;

        // TODO: Execute the join algorithm

        isComplete = true;
        return true;
    }

    bool NestedLoopJoinOperator::execute(int numAvailableBufPages,
                                         File &resultFile) {
        if (isComplete)
            return true;

        numResultTuples = 0;
        numUsedBufPages = 0;
        numIOs = 0;

        // 获得公共属性
        vector <Attribute> common_attributes = getCommonAttributes(leftTableSchema, rightTableSchema);
        // 根据所占页数粗略比较哪个表元组数多
        int left_page_num = 0;
        for (badgerdb::FileIterator iter = leftTableFile.begin(); iter != leftTableFile.end(); ++iter) {
            left_page_num++;
        }
        int right_page_num = 0;
        for (badgerdb::FileIterator iter = rightTableFile.begin(); iter != rightTableFile.end(); ++iter) {
            right_page_num++;
        }
        int min_page_num = left_page_num > right_page_num ? right_page_num : left_page_num;
        TableSchema big_TableSchema =
                left_page_num > right_page_num ? TableSchema(leftTableSchema) : TableSchema(rightTableSchema);
        TableSchema small_TableSchema =
                left_page_num > right_page_num ? TableSchema(rightTableSchema) : TableSchema(leftTableSchema);
        File big_File = left_page_num > right_page_num ? File(leftTableFile) : File(rightTableFile);
        File small_File = left_page_num > right_page_num ? File(rightTableFile) : File(leftTableFile);
        // 计算算法执行过程中使用的缓冲页面数(numUsedBufPages)
        numUsedBufPages = min_page_num >= numAvailableBufPages ? numAvailableBufPages : min_page_num + 1;
        // 嵌套循环算法主体
        badgerdb::FileIterator iter_outer = small_File.begin();
        do {
            // 将外关系的每M-1块读入缓存
            vector <badgerdb::Page> read_pages;
            vector <badgerdb::Page> buffered_pages;
            read_pages.clear();
            buffered_pages.clear();
            for (int i = 0; i < numAvailableBufPages - 1; i++) {
                badgerdb::Page page = *iter_outer;
                badgerdb::Page *buffered_page;
                bufMgr->readPage(&small_File, page.page_number(), buffered_page);
                numIOs++;
                iter_outer++;
                read_pages.push_back(page);
                buffered_pages.push_back(*buffered_page);
                if (iter_outer == small_File.end()) {
                    break;
                }
            }
            // 对每一个内关系的每一块page
            for (badgerdb::FileIterator iter_inner = big_File.begin(); iter_inner != big_File.end(); ++iter_inner) {
                //将page读入缓冲池
                numIOs++;
                badgerdb::Page page = *iter_inner;
                badgerdb::Page *buffered_page;
                bufMgr->readPage(&big_File, page.page_number(), buffered_page);
                // 对page中的每个元组
                for (badgerdb::PageIterator page_iter = buffered_page->begin();
                     page_iter != buffered_page->end(); ++page_iter) {
                    string key_inner = *page_iter;
                    string search_key_inner = construct_search_key(key_inner, common_attributes, big_TableSchema);
                    for (int i = 0; i < read_pages.size(); i++) {
                        Page buf_page = buffered_pages[i];
                        for (badgerdb::PageIterator buf_page_iter = buf_page.begin();
                             buf_page_iter != buf_page.end(); ++buf_page_iter) {
                            // 查找内存中的每个元组
                            string key_outer = *buf_page_iter;
                            string search_key_outer = construct_search_key(key_outer, common_attributes,
                                                                           small_TableSchema);
                            if (search_key_inner.compare(search_key_outer) == 0) {
                                // 连接，并且将结果写入文件
                                string out = "";
                                if (left_page_num > right_page_num) {
                                    out = joinTuples(key_inner, key_outer, big_TableSchema, small_TableSchema);
                                } else {
                                    out = joinTuples(key_outer, key_inner, small_TableSchema, big_TableSchema);
                                }
                                HeapFileManager::insertTuple(out, resultFile, bufMgr);
                                numResultTuples++;
                            }
                        }
                    }
                }
                // 将缓存中的该页面释放
                bufMgr->unPinPage(&big_File, page.page_number(), false);
            }
            // 将缓存中的外关系的所有页面释放
            for (int i = 0; i < read_pages.size(); i++) {
                bufMgr->unPinPage(&small_File, read_pages[i].page_number(), false);
            }
        } while (iter_outer != small_File.end());
        isComplete = true;
        return true;
    }

    BucketId GraceHashJoinOperator::hash(const string &key) const {
        std::hash <string> strHash;
        return strHash(key) % numBuckets;
    }

    bool GraceHashJoinOperator::execute(int numAvailableBufPages,
                                        File &resultFile) {
        if (isComplete)
            return true;

        numResultTuples = 0;
        numUsedBufPages = 0;
        numIOs = 0;

        // TODO: Execute the join algorithm

        isComplete = true;
        return true;
    }

}  // namespace badgerdb